package kafka_topic

import (
	"context"
	"fmt"
	"go-demo-2025/utils/kafkaUtil"
	"log"
	"os/exec"
	"strings"
	"time"
)

// kafka 扩容分区与消费者
// 测试用例: tests/kafka_tests/kafka_topic_test.go

// ResizeKafkaPartitions 适配 kafka-go v0.4.49：通过命令行工具扩容分区
func ResizeKafkaPartitions(ctx context.Context, topic string, newPartitionCount int) error {
	// 1. 基础参数校验
	if newPartitionCount <= 0 {
		return fmt.Errorf("新分区数必须为正整数，当前值: %d", newPartitionCount)
	}
	if len(kafkaUtil.KafkaBrokers) == 0 {
		return fmt.Errorf("Kafka Broker地址未配置（KafkaBrokers为空）")
	}
	if kafkaUtil.KafkaScript == "" {
		return fmt.Errorf("kafka-topics脚本路径未配置（KafkaScript为空）")
	}

	// 2. 查询当前分区数（确保主题存在）
	currentPartitions, err := GetTopicPartitionCount(ctx, topic)
	if err != nil {
		return fmt.Errorf("前置校验失败: %v", err)
	}

	// 3. 扩容逻辑校验
	if newPartitionCount <= currentPartitions {
		return fmt.Errorf("新分区数(%d)必须大于当前分区数(%d)，Kafka不支持分区缩容", newPartitionCount, currentPartitions)
	}

	// 4. 构建并执行扩容命令
	log.Printf("开始扩容主题[%s]：当前分区数=%d → 目标分区数=%d", topic, currentPartitions, newPartitionCount)
	brokerAddr := strings.Join(kafkaUtil.KafkaBrokers, ",") // 支持多Broker集群
	cmd := exec.CommandContext(
		ctx,
		kafkaUtil.KafkaScript,
		"--alter",        // 执行主题修改操作
		"--topic", topic, // 目标主题名
		"--bootstrap-server", brokerAddr, // Broker地址（多地址用逗号分隔）
		"--partitions", fmt.Sprintf("%d", newPartitionCount), // 新分区总数
	)

	// 捕获命令输出（便于排查错误）
	output, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("扩容命令执行失败: 错误=%v, 命令输出=%s", err, string(output))
	}
	log.Printf("扩容命令执行成功，输出: %s", strings.TrimSpace(string(output)))

	// 5. 验证扩容结果（等待Kafka元数据同步）
	log.Printf("等待Kafka元数据同步（1.5秒）...")
	time.Sleep(1500 * time.Millisecond)

	finalPartitions, err := GetTopicPartitionCount(ctx, topic)
	if err != nil {
		return fmt.Errorf("扩容命令执行成功，但验证分区数失败: %v", err)
	}
	if finalPartitions != newPartitionCount {
		return fmt.Errorf("扩容异常：期望分区数=%d，实际分区数=%d", newPartitionCount, finalPartitions)
	}

	log.Printf("主题[%s]分区扩容成功！最终分区数: %d", topic, finalPartitions)
	return nil
}

// GetTopicPartitionCount 适配 kafka-go v0.4.49：查询主题分区数
func GetTopicPartitionCount(ctx context.Context, topic string) (int, error) {
	// 基础校验
	if len(kafkaUtil.KafkaBrokers) == 0 {
		return 0, fmt.Errorf("Kafka Broker地址未配置")
	}
	if kafkaUtil.KafkaScript == "" {
		return 0, fmt.Errorf("kafka-topics脚本路径未配置")
	}

	// 执行查询命令
	brokerAddr := strings.Join(kafkaUtil.KafkaBrokers, ",")
	cmd := exec.CommandContext(
		ctx,
		kafkaUtil.KafkaScript,
		"--describe",
		"--topic", topic,
		"--bootstrap-server", brokerAddr,
	)

	output, err := cmd.CombinedOutput()
	if err != nil {
		return 0, fmt.Errorf("查询命令执行失败: 错误=%v, 输出=%s", err, string(output))
	}

	// 解析输出（适配v0.4.49对应的Kafka版本输出格式）
	outputStr := string(output)
	partitionCountKey := "PartitionCount:"
	if !strings.Contains(outputStr, partitionCountKey) {
		return 0, fmt.Errorf("未找到分区数字段，输出内容: %s", outputStr)
	}

	// 截取并转换分区数（兼容旧版Kafka输出格式）
	parts := strings.SplitAfter(outputStr, partitionCountKey)
	if len(parts) < 2 {
		return 0, fmt.Errorf("解析分区数失败，输出内容: %s", outputStr)
	}
	countStr := strings.TrimSpace(strings.Fields(parts[1])[0]) // 提取第一个字段（分区数）

	var count int
	if _, err := fmt.Sscanf(countStr, "%d", &count); err != nil {
		return 0, fmt.Errorf("转换分区数失败: 原始字符串=%s, 错误=%v", countStr, err)
	}

	log.Printf("主题[%s]当前分区数: %d", topic, count)
	return count, nil
}
